package org.wyz.mapreduce.join;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class JoinMapper extends Mapper<LongWritable, Text, Text, Text> {

    private static final String TABLE_1 = "user.csv";
    private static final String TABLE_2 = "user_exam.csv";


    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        /**
         * 这里我们join两张表
         * 一张是user表：用户表，id都是唯一的
         * 另一张是user_exam表：用户报名考试的relation表，其中的user_id会出现多次
         *
         * map思路：
         * 用join on的id作为key，把剩下的字段作为value，并加上文件名，用来区分这个数据来自哪张表
         *
         * reduce思路：
         * 通过二重循环暴力进行join操作
         *
         */


        // 获取文件名，用来区别数据来自哪张表
        String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();

        // 获取文件绝对路径的写法
        String filePath = ((FileSplit) context.getInputSplit()).getPath().toString();

        List<String> parts = null;

        // 不同csv文件的id位置可能是不一样的
        if (fileName.contains(TABLE_1)) {
            parts = seperate(value.toString(), 0, ",");
        } else if (fileName.contains(TABLE_2)) {
            parts = seperate(value.toString(), 0, ",");
        }

        // 没有我们想要的文件名
        if (parts == null) {
            System.exit(1);
        }

        context.write(new Text(parts.get(0)), new Text(parts.get(1) + "\t" + fileName));
    }

    /**
     * 从行中抽取出id 和 剩下的属性
     *
     * @param row
     * @return 第一个是id，第二个是剩下的attr
     */
    public List<String> seperate(String row, int idPosition, String seperator) {
        String[] attrs = row.split(seperator);
        String id = attrs[idPosition];
        List<String> res = new ArrayList<>();
        res.add(id);
        // 把剩下的拼起来
        StringBuilder sb = new StringBuilder("");
        for (int i = 0; i < attrs.length; i++) {
            if (i == idPosition) {
                continue;
            }
            sb.append(attrs[i]);
            if (i < attrs.length - 1) {
                sb.append(seperator);
            }
        }
        res.add(sb.toString());
        return res;
    }
}
